package com.wuuxiang.polestar.core.zookeeper;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.api.GetDataBuilder;
import org.apache.curator.utils.ZKPaths;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Charsets;
import com.google.common.base.Throwables;
import com.google.common.collect.Maps;
import com.wuuxiang.polestar.core.ConfigGroup;
import com.wuuxiang.polestar.core.GeneralConfigGroup;


public class ZookeeperConfigGroup extends GeneralConfigGroup {
	
	private static final long serialVersionUID = 1L;

	static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperConfigGroup.class);

	// 配置对象
	private ZookeeperConfigProfile configProfile;
	// 节点名称
	private String node;
	// ZK客户端
	private CuratorFramework client;
	// 缓存对象
	private ConfigLocalCache configLocalCache;
	// 节点数据变更监听
	private CuratorListener listener = new ConfigNodeEventListener(this);
	
	public ZookeeperConfigGroup(ZookeeperConfigProfile configProfile, String node) {
		this(null, configProfile, node);
	}
	
	public ZookeeperConfigGroup(ConfigGroup internalConfigGroup,
			ZookeeperConfigProfile configProfile, String node) {
		super(internalConfigGroup);
		this.configProfile = configProfile;
		this.node = node;

		if (configProfile.isOpenLocalCache()) {
			configLocalCache = new ConfigLocalCache(
					FileUtils.getUserDirectoryPath() + "/.polestar",
					configProfile.getRootNode());
		}

		initConfigs();
	}
	
	/**
	 * 初始化节点
	 */
	private void initConfigs() {
		client = CuratorFrameworkFactory.newClient(configProfile.getConnectStr(), configProfile.getRetryPolicy());
		client.start();
		client.getCuratorListenable().addListener(listener);

		LOGGER.info(
				"Loading properties for node: {}, with loading mode: {} and keys specified: {}",
				node, configProfile.getKeyLoadingMode(),
				configProfile.getKeysSpecified());
		loadNode();

		// Update local cache
		if (configLocalCache != null) {
			configLocalCache.saveLocalCache(this, node);
		}
	}
	
	/**
	 * 加载节点并监听节点变化
	 */
	void loadNode() {
		final String nodePath = ZKPaths.makePath(configProfile.getVersionedRootNode(), node);

		final GetChildrenBuilder childrenBuilder = client.getChildren();

		try {
			final List<String> children = childrenBuilder.watched().forPath(nodePath);
			if (children != null) {
				final Map<String, String> configs = Maps.newHashMap();
				for (String child : children) {
					final Pair<String, String> keyValue = loadKey(ZKPaths.makePath(nodePath, child));
					if (keyValue != null) {
						configs.put(keyValue.getKey(), keyValue.getValue());
					}
				}
				cleanAndPutAll(configs);
			}
		} catch (Exception e) {
			LOGGER.error("加载节点并监听异常",e);
			throw Throwables.propagate(e);
		}
	}
	
	void reloadKey(final String nodePath) {
		try {
			final Pair<String, String> keyValue = loadKey(nodePath);
			if (keyValue != null) {
				super.put(keyValue.getKey(), keyValue.getValue());
			} else {
				LOGGER.warn("keyValue为空");
			}
		} catch (Exception e) {
			LOGGER.error("加载变换的key异常:"+e.getMessage(),e);
			throw Throwables.propagate(e);
		}
	}
	
	private Pair<String, String> loadKey(final String nodePath)
			throws Exception {
		final String nodeName = ZKPaths.getNodeFromPath(nodePath);
		final Set<String> keysSpecified = configProfile.getKeysSpecified();
		switch (configProfile.getKeyLoadingMode()) {
		case INCLUDE:
			if (keysSpecified == null || !keysSpecified.contains(nodeName)) {
				return null;
			}
			break;
		case EXCLUDE:
			if (keysSpecified.contains(nodeName)) {
				return null;
			}
			break;
		case ALL:
			break;
		default:
			break;
		}

		final GetDataBuilder data = client.getData();
		final String value = new String(data.watched().forPath(nodePath),
				Charsets.UTF_8);
		return new ImmutablePair<String, String>(nodeName, value);
	}
	
	public String getNode() {
		return node;
	}

	public ConfigLocalCache getConfigLocalCache() {
		return configLocalCache;
	}
	
	/**
	 * 导出属性列表
	 *
	 * @return
	 */
	public Map<String, String> exportProperties() {
		return Maps.newHashMap(this);
	}
	
    @Override
    public void close() throws IOException {
		if (client != null) {
			client.getCuratorListenable().removeListener(listener);
			client.close();
		}
    }
    
}
